using System;
using System.Collections.Generic;
using System.Collections;
using System.Text;
using System.Runtime.Serialization.Formatters.Binary;
using System.IO;
using System.Threading;

namespace ErlyThreads
{
    public class ReceivedMessage
    {
        object mSenderId = null;
        object mData = null;

        public object SenderId
        {
            get { return mSenderId; }
        }

        public object Message
        {
            get { return mData; }
        }

        private ReceivedMessage() { }
        internal ReceivedMessage(
            object pSenderId, 
            object pData)
        {
            mSenderId = pSenderId;
            mData = pData;
        }


        internal static ReceivedMessage CreateMessage(
            object pIdFrom, 
            object pMessage)
        {
            using (MemoryStream stream = new MemoryStream())
            {
                // we have to force the object through serialization
                BinaryFormatter formatter = new BinaryFormatter();

                formatter.Serialize(
                    stream,
                    pMessage);
                stream.Position = 0;
                object message = formatter.Deserialize(stream);

                return new ReceivedMessage(
                    pIdFrom,
                    message);
            }
        }
    }

    public interface IProcessHandle
    {
        void DispatchQueuedMessages(
            Action<ReceivedMessage> pDistributor);
        void DispatchQueuedMessages(
            ICollection<Type> pTypeFilter,
            Action<ReceivedMessage> pDistributor);
        int NumberOfQueuedMessages { get; }
        object Id { get; }
        void SendMessage(
            object pRecipient,
            object pMessage);
        void SubscribeToMessages(
            object pSenderId,
            Converter<Type, bool> pDescriminator);
        void WaitForMessages();
        void WaitForNewMessages();

        /// <summary>
        /// Waits until at least one of each of the type of the messages in the pFilter
        /// is present in the undelivered messages queue.
        /// </summary>
        /// <param name="pFilter"></param>
        void WaitForSpecificMessages(
            ICollection<Type> pFilter);
        void InitProcess(
            object pId,
            Action<IProcessHandle> pStart);
        void InitProcess(
            object pId,
            Action<IProcessHandle> pStart,
            bool pHintIsShortTaks);
        void StartProcess(
            object pId);
    }

    abstract class ThreadContainerBase : IProcessHandle
    {
        struct MessageHandle
        {
            public MessageHandle(
                byte[] pData,
                object pSender)
            {
                mMessageData = pData;
                mSender = pSender;
            }

            public byte[] mMessageData;
            public object mSender;
        }

        /// <summary>
        /// The delegate that the thread will run
        /// </summary>
        protected Action<ThreadContainerBase> mThreadStart;
        /// <summary>
        /// The Id of the current thread.
        /// </summary>
        private object mId;
        private ProcessRuntime mThreadMaster;
        /// <summary>
        /// The queue that will hold the messages. Use the lock to access this variable.
        /// </summary>
        /// <see cref="mMessagesLock"/>
        private Queue<ReceivedMessage> mLockedMessages =
            new Queue<ReceivedMessage>();

        /// <summary>
        /// The lock for the mLockedMessages Queue
        /// </summary>
        readonly object mMessagesLock = new object();

        private IList<AutoResetEvent> mLockedNewMessagesArrivedSignal =
            new List<AutoResetEvent>();
        private readonly object mNewMessagesArrivedSignalLock = new object();

        /// <summary>
        /// Gets the ThreadMaster that handles this ThreadContainer
        /// </summary>
        public ProcessRuntime ThreadMaster
        {
            get { return mThreadMaster; }
        }

        /// <summary>
        /// Gets the Id of the current thread
        /// </summary>
        public object Id
        {
            get { return mId; }
        }

        internal static ThreadContainerBase ThreadContainerBaseFactory(
            ProcessRuntime pMaster,
            Action<IProcessHandle> pStart,
            object pId,
            bool pIsShortTaskHint)
        {
            ThreadContainerBase result = null;

            if (!pIsShortTaskHint)
            {
                result = new ThreadContainer();
            }
            else
            {
                result = new ThreadPoolTypeContainer();
            }
            

            result.mThreadMaster = pMaster;
            result.mId = pId;
            result.mThreadStart = Delegate_Wrap(pStart);

            return result;
        }

        

        /// <summary>
        /// Impliment the thread's start
        /// </summary>
        internal abstract void Start();

        /// <summary>Sends a message to another thread and to the subscribers of this thread's
        /// messages</summary>
        /// <param name="pIdTo">The recipient(s) for the message.</param>
        /// <param name="pMessage">The masseg itself. Will throw if the message is not serializable.</param>
        public void Message_Send(object pIdTo, object pMessage)
        {
            mThreadMaster.Message_Send(
                mId,
                pIdTo,
                pMessage);
        }

        /// <summary>
        /// Subscribe to a filtered list of messages from a named sender
        /// </summary>
        /// <param name="pSenderId">the sender of the message's id</param>
        /// <param name="pFilter">A filter that will determine if you want
        /// to receive messages of a certain type. Pass null to indicate you
        /// are interested in all messages.</param>
        public void Messages_Subscribe(
            object pSenderId,
            Converter<Type, bool> pFilter)
        {
            mThreadMaster.Messages_Subscribe(
                pSenderId,
                Id,
                pFilter != null ?
                    pFilter :
                    delegate(Type t)
                    {
                        return true;
                    });
        }

        /// <summary>
        /// Receives messages to this thread
        /// </summary>
        /// <param name="pIdFrom">The Id of the thread sending the message</param>
        /// <param name="pMessage">The message itself</param>
        internal void Message_Receive(
            ReceivedMessage pReceivedMessage)
        {
            lock (mMessagesLock)
            {
                mLockedMessages.Enqueue(pReceivedMessage);
            }

            ThreadPool.QueueUserWorkItem(delegate(object o)
            {
                lock (mNewMessagesArrivedSignalLock)
                {
                    foreach (AutoResetEvent are in mLockedNewMessagesArrivedSignal)
                    {
                        are.Set();
                    }
                }
            });
        }

        public void DispatchQueuedMessages(
            Action<ReceivedMessage> pCallback)
        {
            Queue<ReceivedMessage>
                newQueue = new Queue<ReceivedMessage>(),
                oldQueue = null;

            // replace the existing queue
            lock (mMessagesLock)
            {
                oldQueue = mLockedMessages;
                mLockedMessages = newQueue;
            }

            // dequeue all the old messages
            while (oldQueue.Count > 0)
            {
                pCallback(oldQueue.Dequeue());
            }
        }

        public void DispatchQueuedMessages(
            ICollection<Type> pMessageTypeFilter,
            Action<ReceivedMessage> pDistributor)
        {
            Queue<ReceivedMessage>
                oldQueue = null,
                deliveredQueue = new Queue<ReceivedMessage>();

            // replace the existing queue
            lock (mMessagesLock)
            {
                oldQueue = mLockedMessages;
                mLockedMessages = new Queue<ReceivedMessage>();

                while (oldQueue.Count > 0)
                {
                    ReceivedMessage temp = oldQueue.Dequeue();

                    if (pMessageTypeFilter.Contains(temp.Message.GetType()))
                    {
                        deliveredQueue.Enqueue(temp);
                    }
                    else
                    {
                        mLockedMessages.Enqueue(temp);
                    }
                }
            }

            while (deliveredQueue.Count > 0)
            {
                pDistributor(deliveredQueue.Dequeue());
            }
        }

        public void WaitForSpecificMessages(
            ICollection<Type> pFilter)
        {
            bool waitSomeMore = true;
            
            Converter<bool, ReceivedMessage[]> getCurrentMessages = 
                delegate(bool pDummy)
                {
                    ReceivedMessage[] result = null;
                    lock(mMessagesLock)
                    {
                        result = new ReceivedMessage[mLockedMessages.Count];
                        mLockedMessages.CopyTo(result, 0);
                    }
                    return result;
                };

            Dictionary<Type, int> exist = new Dictionary<Type, int>(); 
            List<Type> waitingForTypes = new List<Type>(pFilter);
            WaitForMessages();

            while (waitSomeMore)
            {
                foreach (Type t in waitingForTypes)
                {
                    exist[t] = 1;
                }

                foreach (ReceivedMessage o in getCurrentMessages(true))
                {
                    exist[o.Message.GetType()] = 0;
                }

                int count = 0;
                foreach (Type t in waitingForTypes)
                {
                    count += exist[t];
                }

                waitSomeMore = count > 0;
                if (waitSomeMore)
                {
                    WaitForNewMessages();
                }
            }
        }

        /// <summary>
        /// signalled when new messages arrive
        /// </summary>
        private void WaitForNewMessages()
        {
            AutoResetEvent newEvent = new AutoResetEvent(false);
            lock (mNewMessagesArrivedSignalLock)
            {
                mLockedNewMessagesArrivedSignal.Add(newEvent);
            }

            newEvent.WaitOne();

            // lets put freeing up the signal on a background thread
            // so that this one may continue with its little bit of time-slice
            // instead of blocking again
            ThreadPool.QueueUserWorkItem(delegate(object o)
            {
                lock (mNewMessagesArrivedSignalLock)
                {
                    mLockedNewMessagesArrivedSignal.Remove(newEvent);
                }
            });
        }

        /// <summary>
        /// allows the client to determine if any messages have been
        /// queued for receiving.
        /// </summary>
        public int NumberOfQueuedMessages
        {
            get
            {
                int result = 0;
                lock (mMessagesLock)
                {
                    result = mLockedMessages.Count;
                }
                return result;
            }
        }

        /// <summary>
        /// This call blocks until new messages are received.
        /// </summary>
        public void WaitForMessages()
        {
            int count = 0;
            lock (mMessagesLock)
            {
                count = mLockedMessages.Count;
            }

            if (count > 0)
            {
                return;
            }

            WaitForNewMessages();
        }

        /// <summary>
        /// Wraps the thread to handle catered-for eventualities (uncaught exception etc)
        /// </summary>
        static internal Action<ThreadContainerBase> Delegate_Wrap(
            Action<IProcessHandle> pCallback)
        {
            return delegate(ThreadContainerBase pBase)
            {
                try
                {
                    pBase.Message_Send(
                        null,
                        new Messages.ThreadStarted(
                            pBase.Id));

                    pCallback(pBase);
                }
                catch (Exception e)
                {
                    pBase.Message_Send(
                        null,
                        new Messages.ThreadDied(
                            pBase.Id,
                            e));
                }
                finally
                {
                    pBase.Message_Send(
                        null,
                        new Messages.ThreadStopped(
                            pBase.Id));
                }
            };
        }

        #region IProcessHandle Members

        void IProcessHandle.DispatchQueuedMessages(Action<ReceivedMessage> pDistributor)
        {
            this.DispatchQueuedMessages(pDistributor);
        }

        void IProcessHandle.DispatchQueuedMessages(
            ICollection<Type> pTypeFilter,
            Action<ReceivedMessage> pCallback)
        {
            this.DispatchQueuedMessages(
                pTypeFilter,
                pCallback);
        }

        int IProcessHandle.NumberOfQueuedMessages
        {
            get { return this.NumberOfQueuedMessages; }
        }

        object IProcessHandle.Id
        {
            get { return this.Id; }
        }

        void IProcessHandle.SendMessage(object pRecipient, object pMessage)
        {
            this.Message_Send(
                pRecipient,
                pMessage);
        }

        void IProcessHandle.SubscribeToMessages(object pSenderId, Converter<Type, bool> pDescriminator)
        {
            this.Messages_Subscribe(
                pSenderId,
                pDescriminator);
        }

        void IProcessHandle.WaitForMessages()
        {
            this.WaitForMessages();
        }

        void IProcessHandle.InitProcess(
            object pID,
            Action<IProcessHandle> pStart)
        {
            ( this as IProcessHandle ).InitProcess(
                pID,
                pStart,
                false);
        }

        void IProcessHandle.InitProcess(
            object pId,
            Action<IProcessHandle> pStart,
            bool pHintIsShortTask)
        {
            this.mThreadMaster.InitProcess(
                pId,
                pStart,
                pHintIsShortTask);
        }

        void IProcessHandle.StartProcess(
            object pId)
        {
            mThreadMaster.StartProcess(pId);
        }

        void IProcessHandle.WaitForSpecificMessages(
            ICollection<Type> pFilter)
        {
            this.WaitForSpecificMessages(pFilter);
        }

        void IProcessHandle.WaitForNewMessages()
        {
            this.WaitForNewMessages();
        }

        #endregion
    }

    class ThreadContainer : ThreadContainerBase
    {
        internal override void Start()
        {
            Thread theThread = new Thread(delegate()
            {
                this.mThreadStart(this);
            });
            theThread.Start();
        }
    }

    class ThreadPoolTypeContainer : ThreadContainerBase
    {
        internal override void Start()
        {
            ThreadPool.QueueUserWorkItem(delegate(object pDummyState)
            {
                this.mThreadStart(this);
            });
        }
    }
}
